package com.sjgs.gis.service;

import com.google.protobuf.ByteString;
import com.sjgs.common.utils.CompressionUtils;
import com.sjgs.common.utils.DirUtils;
import com.sjgs.common.utils.FileNameUtils;
import com.sjgs.gis.fstrans.*;
import com.sjgs.gis.utils.QETag;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.ClientResponseObserver;
import net.devh.boot.grpc.client.inject.GrpcClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.*;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
/**
 * File client
 * @author jxw
 */
@Service
public class FileService {

    private final Logger logger = LoggerFactory.getLogger(FileService.class);
    private final static int GRPC_CHUNK_SIZE = 2 * 1024 * 1024;

    @Value("${etl.up-dir:/etl/tmp}")
    private String uploadDir;

    @Value("${deploy.download-dir:/deploy/tmp}")
    private String downloadDir;

    @Value("${deploy.app-dir:/usr/local/tomcat/webapps}")
    private String appDir;

    @Value("${deploy.data-dir:/root/data}")
    private String dataDir;

    @Value("${deploy.native-dir:/usr/local/tomcat/native-jni-lib}")
    private String nativeDir;

    @Value("${deploy.plugin-dir:/WEB-INF/lib}")
    private String pluginDir;

    @GrpcClient("geoserver-dc")
    FSTransServiceGrpc.FSTransServiceBlockingStub transServiceBlockingStub;

    @GrpcClient("geoserver-dc")
    FSTransServiceGrpc.FSTransServiceStub transServiceStub;

    /**
     * 下载文件
     *
     * @param fkey
     * @return
     */
    public Path downloadFile(String fkey) {
        String fname = FileNameUtils.getFileName(fkey);
        String fhash = FileNameUtils.getFileHash(fkey);
        String outPath = downloadDir + "/" + fname;
        String unzipOutDir = downloadDir + "/" + fname.substring(0, fname.lastIndexOf("."));
        DirUtils.mkDir(downloadDir);
        FileInfoResponse fileInfoResponse = fetchChunkInfo(fkey);
        long length = fileInfoResponse.getIndexList().size();
        FileChannel outChanel = null;
        FileOutputStream fos = null;
        try {
            fos = new FileOutputStream(outPath);
            outChanel = fos.getChannel();
            for (int i = 0; i < length; i++) {
                long index = fileInfoResponse.getIndex(i);
                Long off = fileInfoResponse.getOffset(i);
                Long len = fileInfoResponse.getLength(i);
                FetchChunkOfFileResponse fetchChunkOfFileResponse = fetchChunkFromFile(fkey, off, len);
                //write chunk
                outChanel.position(off.intValue());
                outChanel.write(ByteBuffer.wrap(fetchChunkOfFileResponse.getChunk().toByteArray()));
                logger.info("fetchChunkFromFile:" + index + ":" + fetchChunkOfFileResponse.getChunk().toByteArray().length);
            }
            if (QETag.calcETag(outPath).equals(fhash)) {
                if (fname.endsWith(".zip") || fname.endsWith(".war")) {
                    try {
                        //解压缩
                        CompressionUtils.zipDecompression(outPath, downloadDir);
                        if (Files.exists(Paths.get(unzipOutDir))) {
                            return Paths.get(unzipOutDir);
                        }
                    } catch (Exception e) {
                        logger.error(e.getMessage());
                    }
                }
            }
        } catch (FileNotFoundException e) {
            logger.error(e.getMessage());
        } catch (IOException e) {
            logger.error(e.getMessage());
        } finally {
            try {
                if (fos != null) fos.close();
                if (outChanel != null) outChanel.close();
            } catch (IOException e) {
                logger.error(e.getMessage());
            }
        }
        return null;
    }

    /**
     * 上传文件
     *
     * @param path
     * @return
     */
    public String uploadFile(String path) {
        try {
            Long start = System.currentTimeMillis();
            File file = new File(path);
            long fsize = file.length();
            String fkey = QETag.calcETag(path) + "@" + file.getName().replaceAll("@", "_");
            MakeFileResponse makeFileResponse = makeFile(path, fkey, fsize, uploadDir, "0");
            if (makeFileResponse.getContext().equals("false")) {
                logger.info("本地文件不存在或者服务器已存在相同文件");
            } else {
                CountDownLatch done = putFile(path, makeFileResponse.getContext());
                //2M=>1s
                long wait_time = Math.round(fsize / GRPC_CHUNK_SIZE);
                if (!done.await(wait_time, TimeUnit.SECONDS)) {
                    logger.warn("async fetchChunk can not finish within " + wait_time + " seconds");
                }
            }
            System.out.println("fetchChunkFromFile cost time :" + String.valueOf((System.currentTimeMillis() - start)) + "ms");
            return fkey;
        } catch (IOException e) {
            ;
        } catch (InterruptedException e) {
            ;
        }
        return null;
    }

    /**
     * 获取文件信息
     */
    private FileInfoResponse fetchChunkInfo(String fkey) {
        FileInfoRequest request = FileInfoRequest.newBuilder()
                .setFkey(fkey)
                .build();
        return transServiceBlockingStub.fetchFileInfo(request);
    }

    /**
     * 下载文件片
     */
    private FetchChunkOfFileResponse fetchChunkFromFile(String context, long off, long len) {
        FetchChunkOfFileRequest request = FetchChunkOfFileRequest.newBuilder()
                .setOffset(off)
                .setLength(len)
                .setContext(context)
                .build();
        return transServiceBlockingStub.fetchChunkFromFile(request);
    }

    /**
     * 创建文件，返回上下文
     *
     * @param path
     * @param fkey
     * @param fsize
     * @param dir
     * @param storeType
     * @return
     * @throws IOException
     */
    private MakeFileResponse makeFile(String path, String fkey, Long fsize, String dir, String storeType) throws IOException {
        if (Files.notExists(Paths.get(path)) || !Files.isReadable(Paths.get(path))) {
            return MakeFileResponse.newBuilder().setContext("false").build();
        }
        MakeFileRequest request = MakeFileRequest.newBuilder()
                .setFkey(fkey)
                .setFsize(fsize)
                .setFdir(dir)
                .setStoretype(storeType)
                .build();
        return transServiceBlockingStub.mkFile(request);
    }

    /**
     * 上传文件
     *
     * @param path
     * @param context
     * @return
     */
    private CountDownLatch putFile(String path, String context) {
        final CountDownLatch done = new CountDownLatch(1);
        if (Files.notExists(Paths.get(path)) || !Files.isReadable(Paths.get(path))) {
            done.countDown();
            return done;
        }
        ClientResponseObserver<PutChunkRequest, PutChunkResponse> responseObserver =
                new ClientResponseObserver<PutChunkRequest, PutChunkResponse>() {
                    ClientCallStreamObserver<PutChunkRequest> requestStream;

                    @Override
                    public void beforeStart(final ClientCallStreamObserver<PutChunkRequest> requestStream) {
                        this.requestStream = requestStream;
                        // @1设置手动流量控制
                        requestStream.disableAutoInboundFlowControl();
                        // @2 当Consumer端有足够空间时自动回调
                        // 序列化protobuf先发送到缓存区（还未到Server端）
                        // Server端需要调用request()向Client拉取消息
                        try {
                            requestStream.setOnReadyHandler(new Runnable() {
                                File file = new File(path);
                                FileInputStream fin = new FileInputStream(file);
                                FileChannel inChannel = fin.getChannel();
                                ByteBuffer byteBuffer = ByteBuffer.allocate(GRPC_CHUNK_SIZE);
                                Long index = 0L;
                                Long offset = 0L;

                                @Override
                                public void run() {
                                    int eof = 0;
                                    while (requestStream.isReady()) {
                                        try {
                                            if ((eof = inChannel.read(byteBuffer)) != -1) {
                                                byteBuffer.flip();
                                                offset += eof;
                                                byte[] bytes = new byte[eof];
                                                byteBuffer.get(bytes);
                                                byteBuffer.clear();
                                                String hash = QETag.calcETag(bytes);
                                                logger.info(index + ":" + hash + ":" + eof);
                                                PutChunkRequest request = PutChunkRequest
                                                        .newBuilder()
                                                        .setContext(context)
                                                        .setHash(hash)
                                                        .setIndex(index)
                                                        .setStart(index * GRPC_CHUNK_SIZE)
                                                        .setEnd(offset)
                                                        .setChunk(ByteString.copyFrom(bytes))
                                                        .build();
                                                // @3 将消息发送到缓存区
                                                requestStream.onNext(request);
                                                index++;
                                            } else {
                                                requestStream.onCompleted();
                                            }
                                        } catch (IOException e) {
                                            e.printStackTrace();
                                        }
                                    }
                                }
                            });
                        } catch (FileNotFoundException e) {
                            e.printStackTrace();
                        }
                    }

                    @Override
                    public void onNext(PutChunkResponse putChunkResponse) {
                        logger.info(putChunkResponse.getChunkId() + " : " + putChunkResponse.getChunkAddr());
                        // @6通知继续发送
                        requestStream.request(1);
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        done.countDown();
                    }

                    @Override
                    public void onCompleted() {
                        done.countDown();
                    }
                };
        transServiceStub.putFile(responseObserver);
        return done;
    }
}
